package com.youth.mq.producer;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

/**
 * @ClassName KafkaProducer
 * @Description: TODO
 * @Author yjwu
 * @Date 2020/4/7 20:54
 **/
@Component
@Slf4j
public class KafkaProducer {
    @Autowired
    private KafkaTemplate kafkaTemplate;

    public void send(String topicName, String key, String jsonMsg) {
        ListenableFuture future = kafkaTemplate.send(topicName, key, jsonMsg);

        future.addCallback(new ListenableFutureCallback<SendResult<String, String>>() {
            @Override
            public void onFailure(Throwable throwable) {
                log.info("kafkaproducer--send--failed-->>{}", throwable.getMessage());
            }

            @Override
            public void onSuccess(SendResult<String, String> result) {
                log.info("kafkaproducer--send--success-->>{}", result.toString());
            }
        });
    }
}
